package cn.com.mqtt.consumer;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import cn.com.gas.service.GasStaytimeService;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class MqttConsumerCallBack implements MqttCallback {

	public static MqttConsumerCallBack mqttConsumerCallBack;
	
	/*
	 * private static GasStaytimeService gasStaytimeService;
	 * 
	 * @Autowired public void setHttpSessionService(GasStaytimeService
	 * gasStaytimeService){ MqttConsumerCallBack.gasStaytimeService =
	 * gasStaytimeService; }
	 */

	private static MqttConsumerConfig client;

	@Autowired
	public void setMqttConsumerConfig(MqttConsumerConfig client) {
		MqttConsumerCallBack.client = client;
	}

	/**
	 * 客户端断开连接的回调
	 */
	@Override
	public void connectionLost(Throwable throwable) {
		log.info("与MQTT服务器断开连接");

		/*if (client.reconnect()) {
			
			log.info("重新连接MQTT成功！");
		} else {
			log.info("重新连接MQTT失败！");
		}*/

	}

	/**
	 * 消息到达的回调
	 */
	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		log.info(String.format("接收消息主题 : %s", topic));
		log.info(String.format("接收消息Qos : %d", message.getQos()));
		String payload = new String(message.getPayload());
		// log.info(String.format("接收消息内容 : %s", payload));
		log.info(String.format("接收消息retained : %b", message.isRetained()));
		// gasStaytimeService.asynSave(payload);
		// 解析
		/*
		 * if (TsBqUtils.isNotNull(payload)) { JSONObject jsonObject =
		 * JSON.parseObject(payload); String from =
		 * String.valueOf(jsonObject.get("from"));
		 * log.info(String.format("from: %s", from)); String time =
		 * String.valueOf(jsonObject.get("time"));
		 * 
		 * JSONArray messageArray =
		 * JSONArray.parseArray(String.valueOf(jsonObject.get("message")));
		 * 
		 * JSONObject messageObject =
		 * JSON.parseObject(String.valueOf(messageArray.get(0))); String
		 * msegType = String.valueOf(messageObject.get("msegType")); String
		 * channel = String.valueOf(messageObject.get("channel")); String
		 * areaNumber = String.valueOf(messageObject.get("areaNumber")); String
		 * areaName = String.valueOf(messageObject.get("areaName")); String
		 * ruleid = String.valueOf(messageObject.get("ruleid")); String ruleName
		 * = String.valueOf(messageObject.get("ruleName")); String prob =
		 * String.valueOf(messageObject.get("prob")); String appear =
		 * String.valueOf(messageObject.get("appear")); String disappear =
		 * String.valueOf(messageObject.get("disappear")); String label =
		 * String.valueOf(messageObject.get("label")); String plateNumber =
		 * String.valueOf(messageObject.get("plateNumber"));
		 * log.info(String.format("label: %s", label)); String image =
		 * String.valueOf(jsonObject.get("image")); }
		 */

		// 异步入库
		// MqttConsumerCallBack.gasStaytimeService.asynSave(payload);
		
		
		
		
		new Thread(() -> {
			GasStaytimeServiceoperate.gasStaytimeServiceoperate.asynSave(payload);
		}).start();

	}

	/**
	 * 消息发布成功的回调
	 */
	@Override
	public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

	}

	// 这是封装 service 插入方法，解决插入数据连接断开
	@Component
	public static class GasStaytimeServiceoperate {
		@Resource
		private GasStaytimeService gasStaytimeService; // 把我们要用到的Dao层操作注入到这个类中
		private static GasStaytimeServiceoperate gasStaytimeServiceoperate; // 必须要重写一下！！

		@PostConstruct
		public void init() {
			gasStaytimeServiceoperate = this;
			gasStaytimeServiceoperate.gasStaytimeService = this.gasStaytimeService; // 改成你们对应的Dao层
		}

		public void asynSave(String payload) { // 把Dao层的操作封装成一个方法

			gasStaytimeService.asynSave(payload);
		}

	}

}